package com.spring.distribute.zk.service;

import com.google.common.collect.Lists;
import com.spring.distribute.DistributeLock;
import com.spring.distribute.common.Constant;
import com.spring.distribute.zk.ZkLockData;
import com.spring.distribute.zk.lcok.ZkDistributeLock;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.TreeCache;
import org.apache.curator.framework.recipes.cache.TreeCacheEvent;
import org.apache.curator.framework.recipes.cache.TreeCacheListener;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;
import org.springframework.beans.factory.SmartInitializingSingleton;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Service;
import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

@Slf4j
@Service
public class ZkService implements SmartInitializingSingleton {
    private static final Integer CREATE_LOOP_COUNT = 5, TREE_CACHE_START_RETRY_COUNT = 5, DELETE_LOOP_COUNT = 5;
    public static final String DISTRIBUTE_LOCK_NODE = "/distribute", SEPARATE = "/",
        CREATE_FAILED = "CREATE_FAILED", DELETE_FAILED = "DELETE_FAILED", INTERRUPT = "INTERRUPT", CHILDREN_NODE_EMPTY = "CHILDREN_NODE_EMPTY";

    @Qualifier("distributeCuratorFramework")
    @Autowired
    private CuratorFramework distributeCurator;

    private static Boolean errorFlag = Boolean.FALSE;

    /**
     * 关闭CuratorFramework
     * */
    public void closeCurator() {
        this.distributeCurator.close();
    }

    public Stat checkIsExistNode(String path) {
        Stat stat = null;
        try {
            stat = this.distributeCurator.checkExists().forPath(path);
        } catch (Exception e) {
            e.printStackTrace();
        }

        return stat;
    }

    public List<String> getChildrens(String path) {
        Integer loop = 0;
        do {
            try {
                return this.distributeCurator.getChildren().forPath(path);
            } catch (Exception e) {
                loop++;

                try {
                    TimeUnit.MILLISECONDS.sleep(100);
                } catch (InterruptedException e1) {
                    return Lists.newArrayList();
                }
                continue;
            }
        } while (loop < TREE_CACHE_START_RETRY_COUNT);

        return Lists.newArrayList();
    }

    public String createNode(String path, CreateMode createMode, String data) {
        String result = CREATE_FAILED;
        Integer loop = 1;
        do {
            try {
                return this.distributeCurator.create().withMode(createMode).forPath(path, data.getBytes(StandardCharsets.UTF_8));
            } catch (Exception e) {
                loop++;

                try {
                    TimeUnit.MILLISECONDS.sleep(100);
                } catch (InterruptedException e1) {
                    // 当前线程如果被中断,则直接返回加锁失败
                    return INTERRUPT;
                }
                continue;
            }
        } while (loop < CREATE_LOOP_COUNT);

        return result;
    }

    public Boolean deleteNode(ZkLockData zkLockData) {
        Boolean delStatus = Boolean.FALSE;
        Integer loop = 0;
        do {
            try {
                Stat stat = this.checkIsExistNode(zkLockData.getCurrentNode());
                if(Objects.isNull(stat)) {
                    return Boolean.TRUE;
                }

                if(Constant.MONITOR_SUCCESS.equals(zkLockData.getCurrentStatus())) {
                    stat = this.distributeCurator.setData().forPath(zkLockData.getCurrentNode(), zkLockData.getMonitorNode().getBytes(StandardCharsets.UTF_8));
                }

                synchronized (ZkDistributeLock.lockSync) {
                    this.distributeCurator.delete().deletingChildrenIfNeeded().forPath(zkLockData.getCurrentNode());
                }
                return Boolean.TRUE;
            } catch (Exception e) {
                loop++;

                try {
                    TimeUnit.MILLISECONDS.sleep(100);
                } catch (InterruptedException e1) {
                    // 当前线程如果被中断,则直接返回加锁失败
                    return delStatus;
                }
                continue;
            }
        } while (loop < DELETE_LOOP_COUNT);

        Executors.newSingleThreadExecutor().submit(() -> {
            do {
                Stat stat = checkIsExistNode(zkLockData.getCurrentNode());
                if(null == stat) {
                    // 异步删除已经不存在，有可能zk问题导致了删除, 比如强制关闭了与本服务的curator关闭了session, 此时直接结束掉当前线程
                    return;
                }

                try {
                    if(Constant.MONITOR_SUCCESS.equals(zkLockData.getCurrentStatus())) {
                        stat = this.distributeCurator.setData().forPath(zkLockData.getCurrentNode(), zkLockData.getMonitorNode().getBytes(StandardCharsets.UTF_8));
                    }

                    this.distributeCurator.delete().deletingChildrenIfNeeded().forPath(zkLockData.getCurrentNode());
                    TimeUnit.MILLISECONDS.sleep(100);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            } while (true);
        });

        return Boolean.FALSE;
    }

    public Object monitorNode(ZkLockData zkLockData) {
        TreeCache treeCache = new TreeCache(this.distributeCurator, zkLockData.getMonitorNode());
        Integer loop = 0;
        Boolean startUp = Boolean.FALSE;
        do {
            try {
                if(!errorFlag && zkLockData.getMonitorNode().endsWith("0000007")) {
                    System.out.println("--> 当前线程 " + Thread.currentThread().getName() + " 监听节点模拟异常");
                    errorFlag = Boolean.TRUE;
                    break;
                }

                // 开启节点监听
                treeCache.start();
                zkLockData.setTreeCache(treeCache);
                DistributeLock.zkLockDataMap.get(zkLockData.getMonitorNode()).setCurrentStatus(Constant.MONITOR_SUCCESS);
                startUp = Boolean.TRUE;
                break;
            } catch (Exception e) {
                loop++;

                try {
                    TimeUnit.MILLISECONDS.sleep(10);
                } catch (InterruptedException e1) {
                    return Boolean.FALSE;
                }
                continue;
            }
        } while (loop < TREE_CACHE_START_RETRY_COUNT);
        if (!startUp) {
            return startUp;
        }

        treeCache.getListenable().addListener(new TreeCacheListener() {
            @Override
            public void childEvent(CuratorFramework curatorFramework, TreeCacheEvent treeCacheEvent) throws Exception {
                if (treeCacheEvent.getType().equals(TreeCacheEvent.Type.NODE_REMOVED)) {
                    zkLockData.setTreeCacheEvent(treeCacheEvent);
                }

                if(treeCacheEvent.getType().equals(TreeCacheEvent.Type.CONNECTION_SUSPENDED) || treeCacheEvent.getType().equals(TreeCacheEvent.Type.CONNECTION_LOST)) {
                    ChildData childData = new ChildData("/NO_PATH", new Stat(), treeCacheEvent.getType().name().getBytes(StandardCharsets.UTF_8));
                    TreeCacheEvent event = new TreeCacheEvent(treeCacheEvent.getType(), childData);
                    zkLockData.setTreeCacheEvent(event);
                }

                synchronized (ZkDistributeLock.lockSync) {
                    ZkDistributeLock.lockSync.notifyAll();
                }
            }
        });

        try {
            ZkDistributeLock.lockSync.wait();

            do {
                if(Objects.isNull(zkLockData.getTreeCacheEvent())) {
                    ZkDistributeLock.lockSync.wait();
                } else {
                    String preThreadMonitorResult = new String(zkLockData.getTreeCacheEvent().getData().getData());
                    if (StringUtils.contains(preThreadMonitorResult, "MONITOR_FAILED_DELETED")) {
                        String preThreadMonitorNode = preThreadMonitorResult.split(",")[0];
                        Stat stat = this.checkIsExistNode(preThreadMonitorNode);
                        if (Objects.isNull(stat)) {
                            return Boolean.TRUE;
                        }

                        ZkLockData beforeZkLockData = DistributeLock.zkLockDataMap.get(preThreadMonitorNode);
                        if (Objects.isNull(beforeZkLockData)) {
                            return Boolean.TRUE;
                        }

                        if (Constant.UN_LOCK.equals(beforeZkLockData.getMonitorNode())) {
                            return Boolean.TRUE;
                        }

                        if (Constant.SUCCESS_CREATE.equals(beforeZkLockData.getMonitorNode()) || Constant.LOCKING.equals(beforeZkLockData.getMonitorNode())) {
                            zkLockData.getTreeCache().close();
                            zkLockData.setTreeCache(new TreeCache(this.distributeCurator, preThreadMonitorNode));
                            try {
                                zkLockData.getTreeCache().start();
                            } catch (Exception e) {
                                return Boolean.FALSE;
                            }

                            beforeZkLockData.setCurrentStatus(Constant.MONITOR_SUCCESS);
                        }
                    } else if(preThreadMonitorResult.equals(TreeCacheEvent.Type.CONNECTION_SUSPENDED.name()) || preThreadMonitorResult.equals(TreeCacheEvent.Type.CONNECTION_LOST.name())) {
                        return Boolean.FALSE;
                    } else {
                        return Boolean.TRUE;
                    }
                }
            } while (true);
        } catch (InterruptedException e) {
            return Boolean.FALSE;
        }
    }

    @Override
    public void afterSingletonsInstantiated() {
        String basePath = ZkService.DISTRIBUTE_LOCK_NODE;
        try {
            this.distributeCurator.delete().deletingChildrenIfNeeded().forPath(basePath);
        } catch (Exception e) {
            e.printStackTrace();
        }

        this.createNode(basePath, CreateMode.PERSISTENT, Thread.currentThread().getName());
    }
}
